package com.dmp.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.dmp.logger.ILogger;

import java.util.Map;

import static com.dmp.storm.base.BaseMapKey.MAPKEY_RESULTMAP_BOLTMSG;
import static com.dmp.storm.base.BaseMapKey.MAPKEY_TUPLEBOLT_RESULTMAP;

/**
 * @author sunzy
 * @version 1.0
 * @date 2016年2月24日 下午2:36:20
 * @desc 数据流为Map<String, Object>流
 */
public abstract class BaseMapBolt extends BaseRichBolt {
    private static final long serialVersionUID = 1L;
    private String keyword;
    protected OutputCollector collector;
    protected ILogger logger;

    @SuppressWarnings({"rawtypes", "unchecked"})
    @Override
    public void execute(Tuple tuple) {
        try {
            Map<String, Object> map = (Map<String, Object>) tuple
                    .getValueByField(MAPKEY_TUPLEBOLT_RESULTMAP);
            Map resultMap = this.process(map);
            if (resultMap != null) {
                collector.emit(tuple, new Values(resultMap, keyword));

            }
            collector.ack(tuple);
        } catch (Exception e) {
            System.err.println(this.getClass().getSimpleName()+ "  BaseMapBolt:execute="+e.getMessage());
            collector.fail(tuple);
        }
    }

    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {
        this.collector = collector;
        prepare();
    }

    public void prepare() {
        //logger = new LocationLogger();
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(MAPKEY_TUPLEBOLT_RESULTMAP, "keyword"));
    }

    public abstract Map<String, Object> process(Map<String, Object> map);

    public void setKeyword(String keyword) {
        this.keyword = keyword;
    }

    public Object getMapBoltMsg(Map<String, Object> map) {
        return map.get(MAPKEY_RESULTMAP_BOLTMSG);
    }

}
